package com.weixin.gong.example.kafka.serializer;

import kafka.serializer.Decoder;
import kafka.serializer.Encoder;
import kafka.utils.VerifiableProperties;

import java.io.*;

/**
 * @author weixin.gong
 * @date 15-7-6 下午4:54
 */
public class JavaSerializer<T> implements Encoder<T>, Decoder<T> {

    public JavaSerializer(VerifiableProperties props) {
        //Encoder必须有这么个构造函数
    }

    public T fromBytes(byte[] bits) {
        if(bits == null || bits.length == 0)
            return null;
        ObjectInputStream ois = null;
        try {
            ByteArrayInputStream bais = new ByteArrayInputStream(bits);
            ois = new ObjectInputStream(bais);
            //noinspection unchecked
            return (T)ois.readObject();
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            if(ois != null)
                try {
                    ois.close();
                } catch (IOException ignored) {}
        }
    }

    public byte[] toBytes(T var) {
        ObjectOutputStream oos = null;
        try {
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            oos = new ObjectOutputStream(baos);
            oos.writeObject(var);
            return baos.toByteArray();
        } catch (Exception e) {
                throw new RuntimeException(e);
        } finally {
            if(oos != null) {
                try {
                    oos.close();
                } catch (IOException ignored) {}
            }
        }
    }
}
